package com.birdy.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.LitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


/**
 * @author birdy
 * @date 2022/4/6 4:07 PM
 */
@Component
@Slf4j
public class FacekillGroupMqFactory implements MqFactory {
    @Autowired
    private MqProperties mqProperties;

    public static final ThreadLocal<MQProducer> PRODUCER_THREAD_LOCAL = new ThreadLocal<>();
    public static final ThreadLocal<LitePullConsumer> PULL_CONSUMER_THREAD_LOCAL = new ThreadLocal<>();
    public static final ThreadLocal<DefaultMQPushConsumer> PUSH_CONSUMER_THREAD_LOCAL = new ThreadLocal<>();

    @Override
    public MQProducer getProducer() throws MQClientException {
        if (PRODUCER_THREAD_LOCAL.get() == null) {
            log.info("no more producer, will make one producer");
            DefaultMQProducer mqProducer = new DefaultMQProducer(mqProperties.getGroup());
            mqProducer.setNamesrvAddr(mqProperties.getNameSrvAddr());
            PRODUCER_THREAD_LOCAL.set(mqProducer);
            mqProducer.start();
            return mqProducer;
        }
        return PRODUCER_THREAD_LOCAL.get();
    }

    @Override
    public LitePullConsumer getLitePullConsumer() {
        if (PULL_CONSUMER_THREAD_LOCAL.get() == null) {
            log.info("no more pull consumer, will make one consumer");
            DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(mqProperties.getGroup());
            consumer.setNamesrvAddr(mqProperties.getNameSrvAddr());
            PULL_CONSUMER_THREAD_LOCAL.set(consumer);
            return consumer;
        }
        return PULL_CONSUMER_THREAD_LOCAL.get();

    }

    @Override
    public DefaultMQPushConsumer getPushConsumer() {
        log.info("no more push consumer, will make one consumer");
        if (PUSH_CONSUMER_THREAD_LOCAL.get() == null) {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqProperties.getGroup());
            consumer.setNamesrvAddr(mqProperties.getNameSrvAddr());
            return consumer;
        }
        return PUSH_CONSUMER_THREAD_LOCAL.get();
    }

}
